package cn.lsh.main;

import cn.lsh.comparator.HotComment2Comparator;
import cn.lsh.comparator.HotComment3Comparator;
import cn.lsh.comparator.HotCommentGroup2Comparator;
import cn.lsh.comparator.HotCommentGroup3Comparator;
import cn.lsh.mapper.HotComment2Mapper;
import cn.lsh.mapper.HotComment3Mapper;
import cn.lsh.mapper.HotCommentMapper;
import cn.lsh.partitioner.HotComment2Partitioner;
import cn.lsh.partitioner.HotCommentPartitioner;
import cn.lsh.reducer.HotComment2Reducer;
import cn.lsh.reducer.HotComment3Reducer;
import cn.lsh.reducer.HotCommentReducer;
import org.apache.commons.lang.ArrayUtils;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;

import java.net.URI;
import java.util.Arrays;

/**
 * 分析所有文件中各个分词占文件的权重：TF-IDF=TF * IDF
 * 词频TF：给定词语在给定文件中出现的次数，一般会做归一化，即除以文件的总词数。tf=n/sum
 * 逆向文件频率IDF：普遍重要性度量，由文件总数除以包含该词的文件的数目，再对商取对数。idf=log(n/sum)
 *
 * 分三步走
 * 1、第一个mapreduce统计总文件数、每个文件的分词数、以及分词后每个词在各文件中出现的次数
 * 2、第二个mapreduce对词频做归一化，并且统计每个词有出现在文件中的文件数目
 * 3、第三个mapreduce计算IDF，再计算各词的tf*idf
 */
public class HotCommentDriver extends BaseDriver {
	public static final String TOTAL_FILE = "part-r-00003";
	public static final String WORD_COUNT_FILE = "part-r-00004";

	public static void main(String[] args) throws Exception {
	    buildJob(true);

	    if ("2".equals(args[0])) {
			step2();
			args[1] = "/test/hot/output";
			// args[1] = "/test/hot/test/";
			args[2] = "/test/hot/output1";
		} else if ("3".equals(args[0])) {
			if (args.length == 1) {
				args[2] = "/test/hot/output2";
			}
			args[1] = "/test/hot/output1";
			step3();
		} else {
	    	step1();
	    	args[2] = "/test/hot/output";
		}
	    args = (String[]) ArrayUtils.subarray(args, 1, 3);
	    setInputAndOutPut(args, true);

	    job.waitForCompletion(true);
	}

	public static void step1() {
		/**
		 *  第一步
		 *  一、mapper通过IKSegmenter分词器对文件进行分词，输出以下三种数据
		 *  	1、word_id 1，文件分的每个词加文件id对应一条记录
		 *  	2、id 1，每个文件id对应一条记录
		 *  	3、count 1，每个文件分的一个词对应一条记录
		 *  二、分区器将数据分为两类：
		 *  	1、word_id 1和word 1分为一类数据
		 *  	2、count 1分为一类
		 *  三、reduce统计数据
		 *  	1、每个词在每个文件出现的次数和每个文件分了多少词
		 *  	2、总的文件数
		 */

		job.setJobName("hot comment");
		//设置格式化类以及分割字符
		job.getConfiguration().set("mapreduce.input.keyvaluelinerecordreader.key.value.separator", " ");
		job.setInputFormatClass(KeyValueTextInputFormat.class);

		job.setMapperClass(HotCommentMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);

		job.setPartitionerClass(HotCommentPartitioner.class);

		//设置4个reduce task，即4个分区
		job.setNumReduceTasks(4);
		job.setReducerClass(HotCommentReducer.class);
	}

	public static void step2() {
		/**
		 *  第二步
		 *  一、mapper对数据映射成三类数据
		 *  二、分区器将数据分成三类，输出到不同结果文件：
		 *  	1、count，文件总数，就一条记录，放在3分区
		 *  	2、词在多少个文件出现的计数，放在4分区
		 *  	3、每个文件分的词数以及原始词在文件中出现的词数两种数据，放在其他3个分区
		 *  三、排序比较器，主要针对第3类数据
		 *  	1、相同文件id的记录排在一起
		 *  	2、id相同的文件的分词数的记录放在最前面
		 *  四、分组比较器
		 *  	1、第一类数据，相同的词放在一组
		 *  	2、第三类数据，相同的文件id放在一组
		 *  五、reduce对数据进行统计
		 *  	1、第一类数据count直接输出
		 *  	2、第二类数据统计出现的文件数输出
		 *  	3、第三类数据，先取第一个文件分词数，然后每个词计算归一化TF输出
		 */
		job.setJobName("hot comment-2");

		job.setInputFormatClass(KeyValueTextInputFormat.class);

		job.setMapperClass(HotComment2Mapper.class);
		job.setOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);

		//设置排序比较器和分组比较器，注意，分组比较器是比较相邻的两个数据，所以需要配合排序使用
		job.setSortComparatorClass(HotComment2Comparator.class);
		job.setGroupingComparatorClass(HotCommentGroup2Comparator.class);

		//设置分区
		job.setPartitionerClass(HotComment2Partitioner.class);
		job.setNumReduceTasks(5);

		job.setReducerClass(HotComment2Reducer.class);

	}

	public static void step3() {
		/**
		 *  第三步，
		 *  需要先计算IDF
		 *  	1、文件总已经有了，在part-r-00003
		 *      2、每个词对应出现的文件的数量也有了part-r-00004
		 *      3、计算公式：idf=log(n/sum)
		 *  一、mapper加载文件数据，计算TF-IDF
		 *  	1、在setUp中加载推送节点本地的part-r-00003和part-r-00004文本里的数据
		 *  	2、计算IDF=log(fileTotal/sum)
		 *  	3、计算TF-IDF=TF*IDF
		 *  二、排序比较器，先根据文件id排序，同id内按TF-IDF值倒序
		 *  三、分组比较器，自定义根据文件id分组
		 *  四、reducer统计每个文件的所有分词
		 */

		job.setJobName("hot comment-3");
		//当客户端在windows启动，程序在liunx运行时，该参数需要设置为true，做格式兼容，默认fase
		job.getConfiguration().set("mapreduce.app-submission.cross-platform", "true");
		//运行平台，这个可以不用设置，默认为yarn
		conf.set("mapreduce.framework.name", "local");
		//集群分布式启动，因为part-r-00003和part-r-00004两个文件数据需要移动到mapper的计算节点
		job.setJar("G:\\bigdata\\hadoop-test\\target\\hadoop-test-1.0-SNAPSHOT.jar");

		//把文件总数加载到job，任务运行时会把该文件推送到计算节点的服务器上
		job.addCacheFile(new Path("/test/hot/output1/" + TOTAL_FILE).toUri());
		//把词对应出现的文件数数据加载到job
		job.addCacheFile(new Path("/test/hot/output1/" + WORD_COUNT_FILE).toUri());

		job.setInputFormatClass(KeyValueTextInputFormat.class);

		job.setMapperClass(HotComment3Mapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);

		job.setSortComparatorClass(HotComment3Comparator.class);
		job.setGroupingComparatorClass(HotCommentGroup3Comparator.class);

		job.setReducerClass(HotComment3Reducer.class);

	}
}
